package com.simon.cdc;

import com.simon.cdc.config.AppConfig;
import com.simon.cdc.config.SourceConfig;
import com.simon.cdc.config.TargetConfig;
import com.simon.cdc.database.DatabaseConnectionConfig;
import com.simon.cdc.serializable.CustomJsonDebeziumDeserializationSchema;
import com.simon.cdc.sink.MySQLDruidSink;
import com.simon.cdc.utils.ParameterParseUtil;
import com.simon.cdc.utils.TableUtils;
import com.simon.cdc.yml.YamlConfigLoader;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.cdc.connectors.mysql.source.MySqlSource;
import org.apache.flink.cdc.connectors.mysql.table.StartupOptions;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.io.IOException;
import java.util.Properties;

public class MySQLToMySQLSync {
    /**
     * 主程序入口
     * 负责初始化任务配置，设置流处理环境，以及构建和执行数据流
     *
     * @param args 命令行参数，通常用于传递配置文件路径等信息
     * @throws Exception 当任务初始化或执行过程中遇到错误时抛出异常
     */
    public static void main(String[] args) throws Exception {
        // 任务初始化，加载job yml配置 和Debezium属性配置
        AppConfig appConfig = loadJobConfig(args);

        Properties properties = loadDebeziumProperties();
        // 配置MySql数据源
        MySqlSource<String> mySqlSource = configurationSource(appConfig, properties);
        // 初始化流执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // 配置流执行环境
        configurationStreamExecutionEnvironment(env, args);
        // 读取数据
        DataStreamSource<String> mysqlDS = env.fromSource(mySqlSource, WatermarkStrategy.noWatermarks(), "mysql-source");
        // 配置MySql Druid Sink
        MySQLDruidSink mySQLDruidSink = configurationSink(appConfig);
        // 将数据流添加到Sink
        mysqlDS.addSink(mySQLDruidSink).name("mysql-sink");
        // 执行环境
        env.execute(appConfig.getFlink().getJob().getName());
    }


    /**
     * 配置 StreamExecutionEnvironment
     * 此方法用于设置执行环境的并行度、重启策略以及检查点（Checkpoint）的相关配置
     *
     * @param env StreamExecutionEnvironment 实例，用于配置执行环境
     */
    private static void configurationStreamExecutionEnvironment(StreamExecutionEnvironment env, String[] args) {
        // 设置并行度
        env.setParallelism(1);
        // 设置重启策略为不重启
        env.setRestartStrategy(RestartStrategies.noRestart());
        // 启用检查点，每5000毫秒（5秒）进行一次
        env.enableCheckpointing(10000);

        configurationCheckpoint(env, args);
    }

    /**
     * 配置检查点的相关设置
     * 此方法主要用于配置Flink作业的检查点路径和相关属性，以支持程序的容错和恢复
     *
     * @param env  StreamExecutionEnvironment的实例，用于配置检查点
     * @param args 命令行参数数组，用于传递检查点路径等配置信息
     */
    private static void configurationCheckpoint(StreamExecutionEnvironment env, String[] args) {
        // 从命令行参数中解析检查点路径，并进行非空校验
        String checkpointPath = ParameterParseUtil.parseArgs(args, "--checkpointPath=")
                .filter(path -> !path.isEmpty())
                .orElse(null);

        // 如果检查点路径有效，则配置检查点存储路径
        if (checkpointPath != null&& !checkpointPath.trim().isEmpty()) {

            env.getCheckpointConfig().setCheckpointInterval(10000);

            env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);

            env.getCheckpointConfig().setCheckpointStorage(checkpointPath);

            // 设置检查点模式为 EXACTLY_ONCE，确保每个数据项仅被处理一次
            env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);

            // 设置外部检查点的清理模式为 RETAIN_ON_CANCELLATION，当作业被取消时保留检查点
            env.getCheckpointConfig().setExternalizedCheckpointCleanup(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
        }
    }


    /**
     * 配置并创建一个MySQLDruidSink对象
     * 该方法根据应用程序配置信息，设置目标数据库的连接配置，并返回一个MySQLDruidSink实例
     * 主要用于在Flink作业中将数据sink到MySQL数据库
     *
     * @param appConfig 应用程序配置对象，包含Flink作业的配置信息
     * @return 返回一个配置好的MySQLDruidSink对象
     */
    private static MySQLDruidSink configurationSink(AppConfig appConfig) {
        // 获取Flink作业的目标数据库配置
        TargetConfig target = appConfig.getFlink().getJob().getTarget();
        // 目标数据库连接配置
        DatabaseConnectionConfig databaseConnectionConfig = new DatabaseConnectionConfig(target.getHost(),
                target.getPort(), target.getDb(), target.getUser(), target.getPwd(), target.getMinPoolSize(), target.getMaxPoolSize());
        // 创建并返回MySQLDruidSink对象
        return new MySQLDruidSink(databaseConnectionConfig);
    }

    /**
     * 创建 MySQL 数据源配置
     *
     * @param appConfig  应用配置对象，包含 Flink 作业的配置信息
     * @param properties 额外的属性配置，用于 Debezium
     * @return 返回一个 MySqlSource 对象，用于读取 MySQL 数据源的数据
     */
    private static MySqlSource<String> configurationSource(AppConfig appConfig, Properties properties) {
        // 获取数据源配置
        SourceConfig source = appConfig.getFlink().getJob().getSource();
        // 构建 MySQL 数据源配置
        return MySqlSource.<String>builder()
                .hostname(source.getHost())  // 设置主机名
                .port(source.getPort())  // 设置端口号
                .databaseList(source.getDb())   // 设置源数据库名
                .tableList(TableUtils.buildTableList(source.getDb(), source.getTables()))    // 设置所有表
                .username(source.getUser())  // 设置用户名
                .password(source.getPwd())  // 设置密码
                .scanNewlyAddedTableEnabled(true)
                .debeziumProperties(getDebeziumConfig(properties))  // 设置 Debezium 属性配置
                .deserializer(new CustomJsonDebeziumDeserializationSchema())  // 设置自定义反序列化器，输出为 JSON 字符串
                .startupOptions(StartupOptions.initial())  // 设置启动选项，初始快照 + 增量
                .build();  // 构建 MySqlSource 对象
    }

    /**
     * 加载作业配置信息
     * 该方法通过YamlConfigLoader工具类初始化配置加载器，并获取配置信息
     * 主要解决了如何从YAML配置文件中加载作业配置的问题
     *
     * @param args 命令行参数，通常用于传递配置文件的路径等信息
     * @return AppConfig对象，包含了从配置文件中加载的作业配置信息
     */
    private static AppConfig loadJobConfig(String[] args) {
        // 初始化配置加载器，此处假设YamlConfigLoader是一个负责读取和解析YAML配置文件的工具类
        YamlConfigLoader.init(args);
        // 获取并返回配置信息对象
        return YamlConfigLoader.getConfig();
    }

    /**
     * 加载Debezium属性配置
     * <p>
     * 该方法从类路径中加载名为"debezium.properties"的配置文件，并将其内容解析到Properties对象中
     * 这些属性配置用于Debezium的转换器设置，以实现MySQL到MySQL的数据同步
     *
     * @return Properties 包含Debezium转换器配置的属性对象
     * @throws IOException 如果无法找到或读取"debezium.properties"文件，则抛出此异常
     */
    private static Properties loadDebeziumProperties() throws IOException {
        Properties debeziumProperties = new Properties();
        // 从类路径加载配置文件并填充Properties对象
        debeziumProperties.load(MySQLToMySQLSync.class.getClassLoader().getResourceAsStream("debezium.properties"));

        return debeziumProperties;
    }


    /**
     * 获取Debezium配置属性
     * 该方法根据传入的转换器属性，构建并返回一个Properties对象，用于配置Debezium
     * 主要配置了时间精度模式、小数处理模式、时区以及时间格式等
     *
     * @param properties 包含转换器相关属性的Properties对象，用于获取时区和格式化信息
     * @return 返回一个配置了Debezium相关属性的Properties对象
     */
    private static Properties getDebeziumConfig(Properties properties) {
        Properties props = new Properties();
        // 基础配置
        props.put("time.precision.mode", "connect");
        props.put("decimal.handling.mode", "string");
        // 时区配置
        props.put("database.serverTimezone", properties.getProperty("timezone"));
        props.put("database.history.timezone", properties.getProperty("timezone"));

        // 转换器配置
        props.put("converters", "timestampConverter");
        props.put("timestampConverter.type", "com.simon.cdc.converters.DateTimeFormatterConverter");
        props.put("timeConverter.datetime.format", "yyyy-MM-dd HH:mm:ss");// 不带毫秒
        // 传递所有以"format."开头的配置项
        properties.stringPropertyNames().stream()
                .filter(key -> key.endsWith("format.") || key.equals("timezone"))
                .forEach(key -> props.put("timestampConverter." + key, properties.getProperty(key)));

        return props;
    }

}

